package com.tsf.hbase.jpa.dao;

import com.tsf.hbase.HBaseClient;
import com.tsf.hbase.RowKey;
import com.tsf.hbase.HBaseOperations;
import com.tsf.hbase.jpa.HBaseColumn;
import com.tsf.hbase.jpa.Page;
import com.tsf.hbase.jpa.HBaseRepository;
import com.tsf.hbase.jpa.HBaseTable;
import com.tsf.hbase.jpa.entity.HBaseEntity;
import com.tsf.hbase.util.BitUtil;
import com.tsf.hbase.util.HBaseUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.filter.FilterList;
import org.springframework.core.annotation.AnnotationConfigurationException;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

/**
 * <p>基于hbase的一些常用操作及自动初始化{@link AbstractHBaseDao}并实现dao层的所有操作选项{@link HBaseDao}的具体实现，子类
 * 可以重写它的所有操作选项，并实现自定义的操作选项逻辑，但必须遵从框架定义的jpa规范，如：dao的类头必须标注{@code @HBaseRepository}。</p>
 * <h2>已实现的操作：</h2>
 * <ul>
 *     <li>自动创建数据表</li>
 *     <li>自动映射实体类</li>
 *     <li>自动创建数据表</li>
 *     <li>新增一条或多条数据</li>
 *     <li>查询一条或所有数据</li>
 *     <li>条件查询多条数据</li>
 *     <li>条件查询多条数据并分页</li>
 * </ul>
 * <h2>一些操作实现的基本原理：</h2>
 * <ul>
 *     <li>1、根据声明实体{@link T}类头标注属性{@link HBaseTable}自动加载数据表名创建hbase表及设置其属性</li>
 *     <li>2、根据{@link T}中的字段标注属性{@link HBaseColumn}自动初始化列属性</li>
 * </ul>
 * <h2>友情提示：</h2>
 * <ul><li>清空hbase数据表可使用命令“truncate 'tableName'”</li></ul>
 * <h2>类使用说明或举例：</h2>
 * <pre>
 *     //定义一个实体类，相当于定义了一个hbase数据表pojo对象
 *     <code>@HBaseTable(name="student", families={"basic-data", "detailed-data", "permissions-data"})</code>
 *     public class Student extends HBaseEntity {
 *         private static final long serialVersionUID = -7993606469720224568L;
 *
 *         <code>@HBaseColumn(family = "basic-data", name = "id")</code>
 *         private String id;
 *
 *         <code>@HBaseColumn(family = "basic-data", name = "name")</code>
 *         private String name;
 *
 *         <code>@HBaseColumn(family = "basic-data", name = "sex")</code>
 *         private String sex;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "grade")</code>
 *         private String grade;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "id_number")</code>
 *         private String idNumber;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "date_birth")</code>
 *         private String dateBirth;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "national")</code>
 *         private String national;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "age")</code>
 *         private int age;
 *
 *         <code>@HBaseColumn(family = "detailed-data", name = "home_address")</code>
 *         private String homeAddress;
 *
 *         <code>@HBaseColumn(family = "permissions-data", name = "login_enable")</code>
 *         private Boolean loginEnable;
 *
 *         //...getter and setter methods
 *     }
 *
 *     //定义一个该实体类的操作选项实现
 *     {@code @HBaseRepository}
 *     public class StudentDao extends{@code BaseHBaseDao<Student>} {
 *         //...你的逻辑实现或自定义dao层操作选项的实现
 *     }
 * </pre>
 *
 * @param <T> 子类声明的实体类型，{@link T}集成至{@link HBaseEntity}是为了约束实体类必须都有一个rowKey属性，便于该类在实现各项操作时使用
 * @author taoningbo
 * @since 1.0, Created as 2018/11/26
 * @see HBaseOperations
 */
public abstract class BaseHBaseDao<T extends HBaseEntity> extends AbstractHBaseDao<T>
        implements HBaseDao<T, RowKey> {

    protected T _get(byte[] rowKey) {
        Get get = new Get(rowKey);
        Result result = hBaseTemplate.get(tableName, get);
        return (result != null && !result.isEmpty()) ? parseResult(result) : null;
    }

    protected Collection<T> _getAll(List<byte[]> rows) {
        List<Get> gets = new ArrayList<>();
        for (byte[] rowKey : rows) {
            gets.add(new Get(rowKey));
        }
        Result[] results = hBaseTemplate.get(tableName, gets);
        if (results != null && results.length > 0) {
            List<T> entities = new ArrayList<>();
            for (Result result : results) {
                if (result != null && !result.isEmpty()) {
                    T entity = parseResult(result);
                    if (entity != null) {
                        entities.add(entity);
                    }
                }
            }
            return entities;
        }
        return Collections.emptyList();
    }

    protected Collection<T> _scan(byte[] rowKey) {
        ResultScanner results = hBaseTemplate.scanner(tableName, rowKey, null);
        if (results != null) {
            try {
                return parseResults(results, null);
            } finally {
                results.close();
            }
        }
        return Collections.emptyList();
    }

    protected Collection<T> _pagingScan(RowKey rowKey, Page page, Long startTime, Long endTime) {
        byte[] startKey;
        if (StringUtils.isNotEmpty(page.getFirstRowKey())) {
            startKey = RowKey.valueOf(page.getFirstRowKey());
        } else {
            startKey = rowKey.setTime(endTime).build();
        }
        byte[] endKey = BitUtil.addBinary(rowKey.setTime(startTime).build(), (byte) 1);
        ResultScanner results = hBaseTemplate.pageScanner(tableName, families, page.getPageSize(),
                startKey, endKey, null, null);
        if (results != null) {
            try {
                return parseResults(results, page);
            } finally {
                results.close();
            }
        }
        return Collections.emptyList();
    }

    protected Collection<T> _pagingFilterScan(T entity, Page page, Long startTime, Long endTime) {
        if (entity == null || (entity.rowKey() == null)) {
            log.warn("The condition object of the query is empty, please fill in the valid condition!");
            return null;
        }
        byte[] startKey;
        if (StringUtils.isNotEmpty(page.getFirstRowKey())) {
            startKey = RowKey.valueOf(page.getFirstRowKey());
        } else {
            startKey = RowKey.buildKey(entity.rowKey(), endTime);
        }
        byte[] endKey = BitUtil.addBinary(RowKey.buildKey(entity.rowKey(), startTime), (byte) 1);
        FilterList filters = getFilterList(entity);
        ResultScanner results = hBaseTemplate.pageScanner(tableName, families, page.getPageSize(),
                startKey, endKey, null, null, filters);
        if (results != null) {
            try {
                return parseResults(results, page);
            } finally {
                results.close();
            }
        }
        return Collections.emptyList();
    }

    /**
     * 初始化一个Dao实例，为了避免子类在实现该抽象类的时候不添加{@link
     * org.springframework.stereotype.Component}注解，导致有些
     * 框架级的类自动注入失败，故强行校验。并只能添加注解类{@link
     * HBaseRepository}。
     */
    public BaseHBaseDao() {
        super();
        if (!checkedRepository(this.getClass())) {
            throw new AnnotationConfigurationException(String.format(
                    "The HBaseDao object '%s' no marked '@HBaseRepository'.",
                    getClass().getName()
            ));
        }
    }

    /**
     * 获取hbase的客户端对象
     *
     * @return 返回初始化的hbase客户端
     */
    @Override
    public final HBaseClient getClient() {
        return hBaseTemplate != null ?
                hBaseTemplate.getHbaseClient() :
                null;
    }

    /* *******************************************************/
    /* *********************   PUT    ************************/
    /* *******************************************************/

    /**
     * 新增一个实体数据.
     *
     * @param entity 待存储的实体对象
     * @return 返回存储结果，如果存储成功则返回‘true’，否则返回‘false’
     */
    @Override
    public final boolean save(T entity) {
        Put put;
        try {
            put = buildPut(entity);
        } catch (IllegalAccessException e) {
            log.error("Builder 'org.apache.hadoop.hbase.client.Put' object error.", e);
            return false;
        }
        return hBaseTemplate.put(tableName, put);
    }

    /**
     * 新增一批实体数据.
     *
     * @param entities 待存储的实体对象集合
     * @return 返回存储结果，如果存储成功则返回‘true’，否则返回‘false’
     */
    @Override
    public final boolean save(Collection<T> entities) {
        List<Put> puts = new ArrayList<>();
        for (T entity : entities) {
            Put put;
            try {
                put = buildPut(entity);
            } catch (IllegalAccessException e) {
                log.warn("Builder 'org.apache.hadoop.hbase.client.Put' object error, discard this data. {}", e.getMessage());
                continue;
            }
            puts.add(put);
        }
        return hBaseTemplate.put(tableName, puts);
    }

    /* *******************************************************/
    /* *********************   GET    ************************/
    /* *******************************************************/

    /**
     * 根据ID查询一条记录.
     *
     * @param rowKey hbase数据表行主键，也是唯一索引
     * @return 返回一个实体查询结果
     */
    @Override
    public final T get(RowKey rowKey) {
        if (rowKey == null) {
            throw new IllegalArgumentException(
                    "You at least specify 'com.tsf.hbase.jpa.RowKey$unique' isn't empty!");
        }
        return _get(rowKey.build());
    }

    /**
     * 根据ID查询一条记录.
     *
     * @param rowKey hbase数据表行主键，也是唯一索引
     * @return 返回一个实体查询结果
     */
    @Override
    public final T get(byte[] rowKey) {
        if (rowKey == null || rowKey.length == 0) {
            throw new IllegalArgumentException("You must specify at least one row index keyword!");
        }
        return _get(rowKey);
    }

    /**
     * 根据ID查询一条记录.
     *
     * @param rowKey hbase数据表行主键，也是唯一索引（byte[]类型的row进行Base64编码后的结果字符串）
     * @return 返回一个实体查询结果
     */
    @Override
    public final T get(String rowKey) {
        if (StringUtils.isEmpty(rowKey)) {
            throw new IllegalArgumentException("You must specify at least one row index keyword!");
        }
        return _get(HBaseUtils.decode(rowKey));
    }

    /**
     * 查询hbase中的所有符合{@code gets}的数据，当数据量较大是一般不建议使用，建议使用分页查询.
     *
     * @param rows rows
     * @return 返回所有符合条件的实体对象集合，不分页，数据量大时不建议使用
     */
    @Override
    public Collection<T> get(List<RowKey> rows) {
        if (rows == null || rows.isEmpty()) {
            throw new IllegalArgumentException(
                    "You at least specify 'com.tsf.hbase.jpa.RowKey$unique' isn't empty!");
        }
        List<byte[]> rowKeys = new ArrayList<>();
        for (RowKey rowKey : rows) {
            if (rowKey != null) {
                rowKeys.add(rowKey.build());
            }
        }
        return _getAll(rowKeys);
    }

    /* *******************************************************/
    /* *********************   SCAN   ************************/
    /* *******************************************************/

    /**
     * 查询hbase中的所有数据，当数据量较大时一般不建议使用，建议使用分页查询.
     *
     * @return 返回hbase表中的所有数据
     */
    @Override
    public final Collection<T> scan() {
        return _scan(null);
    }

    /**
     * 根据指定的{@code rowKey}查询所有匹配到的数据，当数据量较大是一般不建议使用，建议使用分页查询.
     *
     * @param rowKey 实体对象
     * @return 返回所有符合条件的实体对象集合，不分页，数据量大时不建议使用
     */
    @Override
    public final Collection<T> scan(RowKey rowKey) {
        return _scan(rowKey.build());
    }

    /* *******************************************************/
    /* ********************* PAGE SCAN  **********************/
    /* *******************************************************/

    /**
     * 根据指定的{@code rowKey}查询所有匹配到的数据，默认分页查询.
     *
     * @param rowKey 实体对象
     * @param page 分页对象
     * @return 返回所有符合条件的实体对象集合，最多返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public final Collection<T> pagingScan(RowKey rowKey, Page page) {
        return _pagingScan(rowKey, page, null, null);
    }

    /**
     * 根据指定的{@code rowKey}查询所有匹配到的数据并满足开始时间{@code startTime}至当前系统时间之间的数据，默认分页查询.
     *
     * @param rowKey    实体对象
     * @param page      分页对象
     * @param startTime 开始时间
     * @return 返回所有符合条件的实体对象集合，最多返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public Collection<T> pagingScan(RowKey rowKey, Page page, Long startTime) {
//        if (startTime == null) {
//            throw new NullPointerException("startTime");
//        }
        return _pagingScan(rowKey, page, startTime, System.currentTimeMillis());
    }

    /**
     * 根据指定的{@code rowKey}查询所有匹配到的数据并满足开始时间{@code startTime}至结束时间
     * {@code endTime}之间的数据，默认分页查询.
     *
     * @param rowKey    实体对象
     * @param page      分页对象
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @return 返回所有符合条件的实体对象集合，最多返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public Collection<T> pagingScan(RowKey rowKey, Page page, Long startTime, Long endTime) {
//        if (startTime == null) {
//            throw new NullPointerException("startTime");
//        }
        return _pagingScan(rowKey, page, startTime, endTime);
    }

    /* *******************************************************/
    /* **************** PAGE SCAN AND FILTER  ****************/
    /* *******************************************************/

    /**
     * 根据实体中指定的数据查询所有匹配的数据，默认分页查询.
     *
     * @param entity 实体对象
     * @param page   分页对象
     * @return 返回所有符合条件的实体对象集合，最多返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public final Collection<T> pagingFilterScan(T entity, Page page) {
        return _pagingFilterScan(entity, page, null, null);
    }

    /**
     * 根据实体{@code entity}中指定的数据查询所有匹配的数据并满足开始时间{@code startTime}至
     * 当前系统时间之间的数据，默认分页查询.
     *
     * @param entity    实体对象
     * @param page      分页对象
     * @param startTime 开始时间
     * @return 返回所有符合条件的实体对象集合，最后返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public Collection<T> pagingFilterScan(T entity, Page page, Long startTime) {
//        if (startTime == null) {
//            throw new NullPointerException("startTime");
//        }
        return _pagingFilterScan(entity, page, startTime, System.currentTimeMillis());
    }

    /**
     * 根据实体{@code entity}中指定的数据查询所有匹配的数据并满足开始时间{@code startTime}至
     * 结束时间{@code endTime}之间的数据，默认分页查询.
     *
     * @param entity    实体对象
     * @param page      分页对象
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @return 返回所有符合条件的实体对象集合，最后返回{@link Page#DEFAULT_MAX_PAGE_SIZE}的数据，
     * 最少返回{@link Page#DEFAULT_MIN_PAGE_SIZE}
     */
    @Override
    public Collection<T> pagingFilterScan(T entity, Page page, Long startTime, Long endTime) {
        return _pagingFilterScan(entity, page, startTime, endTime);
    }
}
